# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: xguo

import enum
import functools
import hashlib
import json
import logging
import typing
import uuid

import requests
import tabulate
import tornado.gen
import tornado.httpclient
from recordclass import recordclass

import coin.flow.subscriber as flow_subscriber
import coin.flow.topic_map as topic_map
import coin.proto.coin_market_enums_pb2 as coin_enum

from coin.base.config import Config
from coin.base.datetime_util import to_datetime
from coin.base.timestamp import get_timestamp
from coin.exchange.base.order_fill_util import FillManager
from coin.exchange.base.order_gateway_ready import PositionOrderGatewayReady
from coin.exchange.base.rest.task_scheduler import Scheduler, wrap_async_task
from coin.exchange.base.rest_order_gateway_base import PrivateExchangeMessage
from coin.exchange.binance_futures.enum_types import (BinanceFuturesOrderSide,
                                                      BinanceFuturesOrderType,
                                                      BinanceFuturesExecutionType,
                                                      BinanceFuturesOrderStatus)
from coin.exchange.binance_futures.kr_rest.currency import BinanceCurrency
from coin.exchange.binance_futures.kr_rest.futures_product import BinanceFuturesProduct
from coin.exchange.binance.ws.private_subscriber import (BinancePrivateSubscriber,
                                                         BINANCE_FUTURES_WSS_URL)
from coin.exchange.util.rate_limit import RateLimiter
from coin.exchange.util.rest_rate_limit import RestRateLimiter
from coin.proto.coin_order_gateway_pb2 import OrderGatewayConfig, OrderEvent
from coin.proto.coin_query_pb2 import CurrencyBalance, ProductPosition
from coin.strategy.hk_hard_arb.og_util import print_working_orders
from coin.exchange.binance_futures_swap.kr_rest.private_client import BinanceSwapFuturesPrivateParser
from coin.proto.coin_query_pb2 import ProductFillElement
from coin.exchange.binance_futures_swap.kr_rest.native_private_client import (
    BinanceSwapNativePrivateClient)
from coin.exchange.base.rest_order_gateway_base import (
    RestOrderGateway,
    exchange_message_handler,
    create_order_internal,
    BasePositionOrderGatewayInfo,
    to_proto_order_side,
)
from coin.exchange.base.order_gateway import (Order,
                                              OrderSide,
                                              OrderType,
                                              OrderGatewayStatus,
                                              assert_orders_equivalent)


def create_order_from_order_update(message, order_id, timestamp):
  """
  {
    "s":"BTCUSDT",             // Symbol
    "c":"TEST",                // Client Order Id
    "S":"SELL",                // Side
    "o":"LIMIT",               // Order Type
    "f":"GTC",                 // Time in Force
    "q":"0.001",               // Original Quantity
    "p":"9910",                // Price
    "ap":"0",                  // Average Price
    "sp":"0",                  // Stop Price
    "x":"NEW",                 // Execution Type
    "X":"NEW",                 // Order Status
    "i":8886774,               // Order Id
    "l":"0",                   // Order Last Filled Quantity
    "z":"0",                   // Order Filled Accumulated Quantity
    "L":"0",                   // Last Filled Price
    "N": "USDT",               // Commission Asset, will not push if no commission
    "n": "0",                  // Commission, will not push if no commission
    "T":1568879465651,         // Order Trade Time
    "t":0,                     // Trade Id
    "b":"0",                   // Bids Notional
    "a":"9.91"                 // Ask Notional
  }
  """
  field_name_map = {
      "s": "symbol",
      "c": "client_order_id",
      "S": "order_side",
      "o": "order_type",
      "f": "time_in_force",
      "q": "quantity",
      "p": "price",
      "ap": "average_price",
      "sp": "stop_price",
      "x": "execution_type",
      "X": "order_status",
      "i": "order_id",
      "l": "qty_last_fill",
      "z": "qty_accum_fill",
      "L": "price_last_fill",
      "N": "commission_asset",  # not exist when no filled
      "n": "commission_amount",  # not exist when no filled
      "T": "transaction_time",
      "t": "trade_id",
      "b": "bid_notional",
      "a": "ask_notional",
  }
  message = {field_name_map[key]: message.get(key, 0) for key in field_name_map}

  native_symbol = message['symbol']
  product = BinanceFuturesProduct.FromStr(native_symbol)

  order_side = {'BUY': OrderSide.BUY, 'SELL': OrderSide.SELL}[message['order_side']]
  order_type = {'LIMIT': OrderType.LIMIT, 'MARKET': OrderType.MARKET}[message['order_type']]

  price_orig = float(message['price'])
  qty_orig = float(message['quantity'])
  exchange_order_id = str(message['order_id'])
  qty_last_fill = float(message['qty_last_fill'])
  qty_accum_fill = float(message['qty_accum_fill'])
  price_last_fill = float(message['price_last_fill'])
  price_avg_fill = float(message['average_price'])
  fee = -float(message['commission_amount'])

  fill = FillManager(
      price_orig=price_orig,
      price_last_fill=price_last_fill,
      price_avg_fill=price_avg_fill,
      qty_orig=qty_orig,
      qty_accum_fill=qty_accum_fill,
      qty_last_fill=qty_last_fill,
      timestamp_last_fill=timestamp,
  )
  order = Order(order_id=order_id,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=order_side,
                order_type=order_type,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=create_order_internal(
                    timestamp=timestamp,
                    exchange_order_id=exchange_order_id,
                    fill=fill,
                    fee=fee,
                ))
  return order


def create_order_from_open_order_message(msg, timestamp, order_id):
  """
  [
   {
    "symbol": "BTCUSDT",
    "orderId": 1,
    "clientOrderId": "myOrder1",
    "accountId": 1,
    "price": "0.1",
    "origQty": "1.0",
    "cumQty": "1.0",
    "cumQuote": "1.0",
    "status": "NEW",
    "timeInForce": "GTC",
    "type": "LIMIT",
    "side": "BUY",
    "stopPrice": "0.0",
    "updateTime": 1499827319559
    }
  ]
  """
  logger = logging.getLogger(__name__)
  native_symbol = msg['symbol']
  exchange_order_id = str(msg['orderId'])
  product = BinanceFuturesProduct.FromStrNativeProduct(native_symbol)
  price_orig = float(msg['price'])
  qty_orig = float(msg['origQty'])
  qty_accum_fill = float(msg['executedQty'])
  amount_accum_fill = float(msg['cumQuote'])

  if qty_accum_fill > 1.e-10:
    price_avg_fill = amount_accum_fill / qty_accum_fill
  elif qty_accum_fill < 0:
    logger.error('Insane qty_accum_fill: %s', qty_accum_fill)
    qty_accum_fill = 0
    price_avg_fill = 0
  else:
    qty_accum_fill = 0
    price_avg_fill = 0

  order_type = OrderType.LIMIT
  order_side = OrderSide.BUY if msg['side'] == 'BUY' else OrderSide.SELL

  fill = FillManager(price_orig=price_orig,
                     qty_orig=qty_orig,
                     price_last_fill=None,
                     qty_last_fill=None,
                     price_avg_fill=price_avg_fill,
                     qty_accum_fill=qty_accum_fill,
                     timestamp_last_fill=timestamp)

  order = Order(order_id=order_id,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=order_side,
                order_type=order_type,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=create_order_internal(timestamp=timestamp,
                                               exchange_order_id=exchange_order_id,
                                               fill=fill))
  return order


class BinanceSwapFuturesOrderGatewayInfo(BasePositionOrderGatewayInfo):
  CurrencyType = BinanceCurrency
  ProductType = BinanceFuturesProduct
  BalanceType = float
  PositionType = float

  def __init__(self,
               currencies: typing.List[BinanceCurrency],
               products: typing.List[BinanceFuturesProduct]):
    super().__init__(currencies, products)
    self._unrealized_pnl = None
    self.set_zero_positions()
    self.set_zero_balances()
    self._avg_entry_price = {product: None for product in self.products}

  def update_balance_rest(self, message):
    account_balance = BinanceSwapFuturesPrivateParser.parse_balance(message)
    for balance in account_balance.each_balance:
      currency = BinanceCurrency.FromStrCurrency(balance.currency)
      self.set_balance(currency, balance.total, balance)
    self._unrealized_pnl = BinanceSwapFuturesPrivateParser.parse_unrealized_pnl(message)

  def update_position_rest(self, message):
    account_position = BinanceSwapFuturesPrivateParser.parse_account_position(message)
    for position in account_position.each_position:
      try:
        product = BinanceFuturesProduct.FromStr(position.symbol)
      except Exception:
        continue
      self.set_position(product, position.net_position, position)
      self._avg_entry_price[product] = float(position.avg_entry_price)

  def update_balance_position_ws(self, message):
    for balance in message['a']['B']:
      currency = BinanceCurrency.FromStrNativeCurrencyNothrow(balance['a'])  # asset
      if currency is None:
        continue
      total = float(balance['wb'])
      assert total >= 0.0
      currency_balance = CurrencyBalance(
          currency=currency.currency,
          currency_native=currency.native_currency,
          total=total,
      )
      self.set_balance(currency, currency_balance.total, currency_balance)

    for position in message['a']['P']:
      product = BinanceFuturesProduct.FromStr(position['s'])
      product_position = ProductPosition(
          symbol=product.symbol,
          net_position=float(position['pa']),
      )
      self.set_position(product, product_position.net_position, product_position)
      self._avg_entry_price[product] = float(position['ep'])

  # Margin balance includes unrealized pnl, but this is provided by exchange
  # and could be out of date or slow.
  # DO NOT USE THIS IN TRADING, ONLY ACCEPTABLE IN PNL DISPLAY.
  def get_margin_balance(self, currency: str):
    self.get_balance(currency) + float(self._unrealized_pnl[currency])

  def get_average_entry_price(self, product):
    return self._avg_entry_price.get(product, None)


OrderStats = recordclass('OrderStats',
                         [
                             'num_submitted',
                             'num_cancelled',
                             'num_submitted_success',
                             'num_cancelled_success',
                             'num_fills',
                             'disabled',
                             'disabled_timestamp',
                             'conversion_ratio',
                         ])


# For Binance's rate limiting rule, please refer to
# https://support.binance.com/hc/en-us/articles/360004492232-API-Frequently-Asked-Questions-FAQ
class BinanceSwapFuturesOrderGateway(RestOrderGateway):
  _exchange = 'Binance'
  _market_type = 'Futures'
  _api_version = None

  OGInfoType = BinanceSwapFuturesOrderGatewayInfo

  class _MsgType(enum.IntEnum):
    WS_ORDER_TRADE_UPDATE = 10
    WS_ACCOUNT_UPDATE = 11

    REST_ACCOUNT_BALANCE = 20
    REST_ACCOUNT_POSITION = 21
    REST_OPEN_ORDERS = 22

  def __init__(self,
               currencies: typing.List[BinanceCurrency],
               products: typing.List[BinanceFuturesProduct],
               config: Config,
               og_config=None,
               logger=None):
    super().__init__(currencies, products, config, og_config, logger)
    self._og_ready = PositionOrderGatewayReady()
    self._og_ready.set_status(OrderGatewayStatus.INITIALIZING)
    self._exchange_info = None

    self._private_client = None
    self._private_subscriber = None
    self._scheduler = None

    self._ioloop = None
    self._flow_sub = None
    self._publisher = None

    default_og_config = OrderGatewayConfig(use_fill_checker=True,
                                           account_query_period=8,
                                           user_query_period=2,
                                           orders_snapshot_query_period=4,
                                           order_rate_limit=10,
                                           order_rate_limit_per=1,
                                           hit_rate_limit_cooldown_time=300)
    default_og_config.MergeFrom(self._og_config)
    self._og_config = default_og_config
    self._og_logger.exchange_type = coin_enum.Binance

    self._rest_rate_limiter = RestRateLimiter(delay=self._og_config.hit_rate_limit_cooldown_time)
    self._order_rate_limiter = RateLimiter(self._og_config.order_rate_limit,
                                           self._og_config.order_rate_limit_per)
    self._og_logger.exchange_type = coin_enum.Binance

    # for client order id
    self._uuid = 'presto' + hashlib.md5(str(uuid.uuid1()).encode()).hexdigest()[:16]

    self._order_stats = {}
    for product in self._og_info.products:
      self._order_stats[product] = OrderStats(
          num_submitted=0,
          num_cancelled=0,
          num_submitted_success=0,
          num_cancelled_success=0,
          num_fills=0,
          disabled=False,
          disabled_timestamp=None,
          conversion_ratio=0,
      )

  def _to_client_order_id(self, order):
    client_order_id = '%s_%s' % (self._uuid, order.order_id)
    return client_order_id

  def _parse_client_order_id(self, client_order_id):
    idx = client_order_id.find('_')
    if idx == -1:
      return None, None

    og_uuid = client_order_id[:idx]
    if og_uuid != self._uuid:
      self._logger.info('External order %s, self_og_id=%s', client_order_id, self._uuid)
      return None, None

    order_id = client_order_id[idx + 1:]
    return uuid, order_id

  def _disable_og(self):
    self._scheduler.stop()
    self._private_subscriber.stop()
    self._og_ready.reset()

  def _handle_error_response(self, response, extra_params=None):
    if response.status_code == requests.codes.unauthorized:  # 401
      # {"code": -2015, "msg": "Invalid API-key, IP, or permissions for action."}
      self._logger.error('Conversion ratio is too low!')
      extra_params = extra_params or {}
      order = extra_params.get('order')
      if order is not None:
        self._order_stats[order.product].disabled = True
        self._order_stats[order.product].disabled_timestamp = get_timestamp()
      self._rest_rate_limiter.set_error(status_code=response.status_code)
    elif response.status_code in (requests.codes.teapot, requests.codes.too_many_requests):
      # (418, 420) rate limit exceeded
      self._logger.error('(%s) Hit rate limit, slow down!', response.status_code)
      self._on_hit_order_rate_limit()
    elif response.status_code == requests.codes.forbidden:  # 403:
      self._logger.error('Violate WAF rule!')
      self._on_hit_order_rate_limit()
    else:
      self._logger.error('Response error: %s, %s', response.status_code, response.content)

  def start(self, queue, ioloop, flow_sub=None):
    api_host = self._og_config.rest_root_url if self._og_config.HasField('rest_root_url') else None
    self._private_client = BinanceSwapNativePrivateClient(key_file=self._config.key_file,
                                                          api_host=api_host,
                                                          use_async=True)

    topic_id = topic_map.search_topic_id('binance_rest_private_1')[0][1]
    self._ioloop = ioloop
    self._publisher = queue.get_writer(topic_id=topic_id)
    self._scheduler = Scheduler(ioloop=ioloop, logger=self._logger)
    self._prepare_private_subscriber()
    self._install_periodic_query()

    # Subscribe to flow for feed.
    self._flow_sub = flow_sub or flow_subscriber.from_queue(queue)
    self._flow_sub.subscribe('binance_rest_private_1', self.on_private_exchange_msg)
    self._get_exchange_info()

  def _prepare_private_subscriber(self, start=False):
    if self._private_subscriber is not None:
      self._private_subscriber.stop()

    self._private_subscriber = BinancePrivateSubscriber(
        self._private_client,
        self._ioloop,
        self._notification_callback,
        functools.partial(self._prepare_private_subscriber, start=True),
        300,
        BINANCE_FUTURES_WSS_URL)

    if start:
      self._private_subscriber.start()

  def submittable(self):
    is_check = self._order_rate_limiter.check()
    is_avail = self._rest_rate_limiter.is_available()
    return (self.is_ready() and is_check and is_avail)

  def get_order_stats(self):
    self._calc_conversion_ratio()
    return self._order_stats

  def get_average_entry_price(self, product):
    return self._og_info.get_average_entry_price(product)

  def _calc_conversion_ratio(self):
    for product in self._og_info.products:
      total = self._order_stats[product].num_submitted + self._order_stats[product].num_cancelled
      if total > 0:
        conversion_ratio = self._order_stats[product].num_fills / total
      else:
        conversion_ratio = 0
      self._order_stats[product].conversion_ratio = conversion_ratio

  @functools.lru_cache(maxsize=1024)
  def get_min_qty(self, product):
    for symbol_info in self._exchange_info['symbols']:
      if symbol_info['symbol'] == product.native_symbol:
        for filter_info in symbol_info['filters']:
          if filter_info['filterType'] == 'LOT_SIZE':
            return float(filter_info['minQty'])
    return None

  @tornado.gen.coroutine
  def _get_exchange_info(self):
    client = tornado.httpclient.AsyncHTTPClient()
    response = yield client.fetch('https://fapi.binance.com/fapi/v1/exchangeInfo')
    if response.code == requests.codes.ok:
      self._exchange_info = json.loads(response.body)
      self._logger.info('Exchange Info retrieved!')
      self.start_scheduler()
      self._private_subscriber.start()
    else:
      self._logger.error('Failed to get exchange info!')
      self._ioloop.call_later(5, self._get_exchange_info)

  @tornado.gen.coroutine
  def _on_hit_order_rate_limit(self):
    if not self.is_ready():
      return

    self._scheduler.stop()
    self._og_ready.reset()
    self._logger.info('Hit rate limit, disable OG for % seconds.',
                      self._og_config.hit_rate_limit_cooldown_time)
    yield tornado.gen.sleep(self._og_config.hit_rate_limit_cooldown_time)
    self._scheduler.start()
    self._logger.info('Binance OG is reinitializing.')

  @tornado.gen.coroutine
  def _install_periodic_query(self):
    assert self._scheduler is not None, 'self._scheduler is not set!'
    assert self._publisher is not None, 'self._publisher is not set!'

    # query balance
    query_balance_task = wrap_async_task(
        self._private_client.query_account,
        msg_type=self._MsgType.REST_ACCOUNT_BALANCE,
        publisher=self._publisher,
    )

    self._scheduler.add_periodic_task(
        period=self._og_config.account_query_period,
        job_func=query_balance_task,
    )

    # query position
    query_position_task = wrap_async_task(
        self._private_client.query_position_risk,
        msg_type=self._MsgType.REST_ACCOUNT_POSITION,
        publisher=self._publisher,
    )

    self._scheduler.add_periodic_task(
        period=self._og_config.user_query_period,
        job_func=query_position_task,
    )

    # query active orders
    for product in self._og_info.products:
      query_orders_task = wrap_async_task(
          self._private_client.query_open_orders,
          msg_type=self._MsgType.REST_OPEN_ORDERS,
          publisher=self._publisher,
          func_params=dict(symbol=product.native_symbol),
      )

      yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)
      self._scheduler.add_periodic_task(
          period=self._og_config.orders_snapshot_query_period,
          job_func=query_orders_task,
      )

    def print_info():
      print_working_orders(self._logger, self)
      self._calc_conversion_ratio()
      headers = [
          'product',
          'num_submitted',
          'num_submitted_success',
          'num_cancelled',
          'num_cancelled_success',
          'num_fills',
          'disabled',
          'disabled_timestamp',
          'conversion_ratio'
      ]
      table = []
      for prod, stat in self._order_stats.items():
        if stat.disabled_timestamp:
          disabled_time = to_datetime(stat.disabled_timestamp).strftime('%Y%m%d %H%M%S')
        else:
          disabled_time = 'N/A'
        table.append([
            prod.symbol,
            stat.num_submitted,
            stat.num_submitted_success,
            stat.num_cancelled,
            stat.num_cancelled_success,
            stat.num_fills,
            stat.disabled,
            disabled_time,
            stat.conversion_ratio,
        ])
      fmt_table = tabulate.tabulate(table, headers, stralign="right")
      self._logger.info('----- Binance conversion ratio\n%s', fmt_table)

    # Periodically print working order list.
    self._scheduler.add_periodic_task(period=10, job_func=print_info)

  def _async_submit_order(self, order, post_only=False):
    assert order.order_id is not None, order
    assert order.order_type == OrderType.LIMIT, order

    if self._order_stats[order.product].disabled:
      self._logger.warning('Product %s has low conversion ratio, it is disabled!')
      return

    min_qty = self.get_min_qty(order.product)
    if order.qty < min_qty:
      self._order_manager.remove_order(order)
      self._logger.error('%s: qty (%s) < min_qty (%s)', order.product.symbol, order.qty, min_qty)
      return

    client_order_id = self._to_client_order_id(order)
    formatter = self._order_formatter[order.product]
    binance_order_type = BinanceFuturesOrderType(order.order_type.name)
    params = {
        'symbol': order.product.native_symbol,
        'price': formatter.format_price(order.price),
        'quantity': formatter.format_qty(order.qty),
        'order_side': BinanceFuturesOrderSide(order.order_side.name),
        'order_type': binance_order_type,
        'new_client_order_id': client_order_id,
    }
    fut = self._private_client.place_order(**params)
    self._order_stats[order.product].num_submitted += 1
    return fut

  def _async_cancel_order(self, order):
    assert order.internal.exchange_order_id is not None, order
    fut = self._private_client.cancel_order(symbol=order.product.native_symbol,
                                            order_id=order.internal.exchange_order_id)
    self._order_stats[order.product].num_cancelled += 1
    return fut

  def _handle_submit_response(self, order, response):
    """
    {
      "accountId": 10012,
      "clientOrderId": "testOrder",
      "cumQuote": "0",
      "executedQty": "0",
      "orderId": 22542179,
      "origQty": "10",
      "price": "10000",
      "side": "SELL",
      "status": "NEW",
      "stopPrice": "0",
      "symbol": "BTCUSDT",
      "timeInForce": "GTC",
      "type": "LIMIT",
      "updateTime": 1566818724722
    }
    // This error
    {'code': -1013, 'msg': 'Filter failure: PRICE_FILTER'}
    """
    msg = response.json()
    if response.status_code != requests.codes.ok:
      self._logger.error('Order submission error: %s', msg)
      if order in self._order_manager:
        self._order_manager.remove_order(order, ignore_error=True)
      self._handle_error_response(response)
      return OrderEvent.ORDER_REJECTED

    self._order_stats[order.product].num_submitted_success += 1
    client_order_id = msg['clientOrderId']
    if not self._parse_client_order_id(client_order_id)[1] == order.order_id:
      self._logger.error('Weird! Wrong client_order_id: %s, %s', msg, order)
      self._order_manager.remove_order(order, ignore_error=True)
      return OrderEvent.ORDER_ERROR

    exchange_order_id = str(msg['orderId'])
    prev_order = self._order_manager.get(exchange_order_id=exchange_order_id)
    if prev_order is None:
      self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
    else:
      if prev_order.order_id != order.order_id:
        self._logger.info('order id replaced: %s => %s', prev_order.order_id, order.order_id)
        self._order_manager.remove_order(prev_order)
        self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
        self._order_manager.remove_order(order)
        prev_order.order_id = order.order_id
        self._order_manager.add_order(prev_order)

    status = msg.get('status', None)
    if status is not None:
      try:
        if BinanceFuturesOrderStatus(status) != BinanceFuturesOrderStatus.NEW:
          self._logger.info('submit response: %s', msg)
      except ValueError:
        self._logger.error('Unknown order status: %s', msg)
    else:
      self._logger.debug('Weird, status field is missing: %s', msg)
    return OrderEvent.ORDER_ACCEPTED

  def _handle_cancel_response(self, order, response):
    """
    {
      "symbol": "BTCUSDT",
      "orderId": 28,
      "origClientOrderId": "myOrder1",
      "clientOrderId": "cancelMyOrder1",
      "transactTime": 1507725176595,
      "price": "1.00000000",
      "origQty": "10.00000000",
      "executedQty": "8.00000000",
      "cumQuote": "8.00000000",
      "status": "CANCELED",
      "timeInForce": "GTC",
      "type": "LIMIT",
      "side": "SELL"
     }
    """
    msg = response.json()
    if response.status_code != requests.codes.ok:
      self._logger.error('Cancel order error! %s\n%s', msg, order)
      self._handle_error_response(response)
      if msg['code'] == -2011 and msg['msg'] == "Unknown order sent.":
        if not self._order_manager.is_dead_order(order):
          self._logger.info('Cannot found order!, treat as cancel confirmed!')
          self._order_manager.set_order(order, cancel_confirmed=True)
      return OrderEvent.CANCEL_ERROR

    if BinanceFuturesOrderStatus(msg['status']) != BinanceFuturesOrderStatus.CANCELED:
      self._logger.error('Cancel response %s', msg)
      return OrderEvent.CANCEL_ERROR

    if str(msg['orderId']) != order.internal.exchange_order_id:
      self._logger.error('Inconsistency order id: %s, %s', msg, order)
      return OrderEvent.CANCEL_ERROR

    self._order_stats[order.product].num_cancelled_success += 1
    if not self._order_manager.is_dead_order(order):
      self._logger.debug('%s is cancel confirmed.', order.order_id)
      self._order_manager.set_order(order, cancel_confirmed=True)
    else:
      self._logger.debug('Order is dead already %s', order)
    return OrderEvent.CANCEL_CONFIRMED

  def _notification_callback(self, data):
    timestamp = get_timestamp()
    event_to_msg_type = {
        'ORDER_TRADE_UPDATE': self._MsgType.WS_ORDER_TRADE_UPDATE,
        'ACCOUNT_UPDATE': self._MsgType.WS_ACCOUNT_UPDATE,
    }

    msg_type = event_to_msg_type.get(data['e'], None)
    if msg_type is None:
      self._logger.warning('Unknown event type: %s', data)
      return

    exchange_message = PrivateExchangeMessage(msg_type=msg_type,
                                              data=data,
                                              received_timestamp=timestamp,
                                              request_params=None,
                                              request_timestamp=None)
    self._publisher.write(timestamp, exchange_message)

  @exchange_message_handler(_MsgType.REST_ACCOUNT_BALANCE)
  def _handle_rest_account_balance(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest open orders response error! %s, %s',
                         response.status_code,
                         response.content)
      self._handle_error_response(response)
      return

    message = response.json()
    self._og_info.update_balance_rest(message)
    self._og_logger.write_balance(self._og_info.balance_as_proto())
    if not self.is_ready():
      self._og_ready.set_balance_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.REST_ACCOUNT_POSITION)
  def _handle_rest_account_position(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest open orders response error! %s, %s',
                         response.status_code,
                         response.content)
      self._handle_error_response(response)
      return

    message = response.json()
    self._og_info.update_position_rest(message)
    self._og_logger.write_position(self._og_info.position_as_proto())
    if not self.is_ready():
      self._og_ready.set_position_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.REST_OPEN_ORDERS)
  def _handle_rest_open_orders(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    timestamp = exchange_message.received_timestamp
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest open orders response error! %s, %s',
                         response.status_code,
                         response.content)
      self._handle_error_response(response)
      return

    message = response.json()
    exchange_open_order_dict = {}
    products = set()
    for order_msg in message:
      native_symbol = order_msg['symbol']
      product = BinanceFuturesProduct.FromStrNativeProduct(native_symbol)
      if product not in self._og_info.products:
        continue

      products.add(product)
      order = create_order_from_open_order_message(order_msg, timestamp, None)
      order_id = self._parse_client_order_id(order_msg['clientOrderId'])[1]
      if order_id is None:
        prev_order = self._order_manager.get(exchange_order_id=order.internal.exchange_order_id)
        if prev_order:
          order_id = prev_order.order_id
        else:
          order_id = self._order_id_generator.gen_order_id(is_foreign=True)
      order.order_id = order_id
      exchange_open_order_dict[order.internal.exchange_order_id] = order

    self._order_manager.copy_from_exchange_open_orders(exchange_open_order_dict, products)

    if not self.is_ready():
      self._og_ready.set_orders_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.WS_ACCOUNT_UPDATE)
  def _handle_ws_account_update(self, exchange_message: PrivateExchangeMessage):
    pass
    # temporary disable ws position update
    # message = exchange_message.data
    # self._og_info.update_balance_position_ws(message)
    # self._og_logger.write_balance(self._og_info.balance_as_proto())
    # self._og_logger.write_position(self._og_info.position_as_proto())

  @exchange_message_handler(_MsgType.WS_ORDER_TRADE_UPDATE)
  def _handle_ws_order_update(self, exchange_message: PrivateExchangeMessage):
    """
    {
      "e":"ORDER_TRADE_UPDATE",    // Event Type
      "E":1568879465651,           // Event Time
      "o":{
        "s":"BTCUSDT",             // Symbol
        "c":"TEST",                // Client Order Id
        "S":"SELL",                // Side
        "o":"LIMIT",               // Order Type
        "f":"GTC",                 // Time in Force
        "q":"0.001",               // Original Quantity
        "p":"9910",                // Price
        "ap":"0",                  // Average Price
        "sp":"0",                  // Stop Price
        "x":"NEW",                 // Execution Type
        "X":"NEW",                 // Order Status
        "i":8886774,               // Order Id
        "l":"0",                   // Order Last Filled Quantity
        "z":"0",                   // Order Filled Accumulated Quantity
        "L":"0",                   // Last Filled Price
        "N": "USDT",               // Commission Asset, will not push if no commission
        "n": "0",                  // Commission, will not push if no commission
        "T":1568879465651,         // Order Trade Time
        "t":0,                     // Trade Id
        "b":"0",                   // Bids Notional
        "a":"9.91"                 // Ask Notional
      }
    }
    """
    message = exchange_message.data['o']
    timestamp = exchange_message.received_timestamp
    execution_type = BinanceFuturesExecutionType(message['x'])
    client_order_id = message['c']
    trade_id = message['t']

    order_id = self._parse_client_order_id(client_order_id)[1]
    recv_order = create_order_from_order_update(message, order_id, timestamp)
    if recv_order.product not in self._og_info.products:
      return

    exchange_order_id = recv_order.internal.exchange_order_id
    work_order = self._order_manager.get(exchange_order_id=exchange_order_id)

    # In case a foreign order that has not been added yet
    if order_id is None and work_order is None:
      order_id = self._order_id_generator.gen_order_id(is_foreign=True)
      recv_order.order_id = order_id
      self._order_manager.add_order(recv_order)
      self._logger.info('Add foreign order: %s', recv_order)

    order = self._order_manager.get(exchange_order_id=exchange_order_id)
    if order is None:
      order = self._order_manager.get(order_id=recv_order.order_id)
      if order is None:
        self._order_manager.add_order(recv_order)
        order = self._order_manager.get(order_id=recv_order.order_id)

    if order:
      try:
        assert_orders_equivalent(order, recv_order)
      except AssertionError:
        self._logger.warning('assertion order equivalent failed! %s\n%s', order, recv_order)
    else:
      self._logger.error('Unknown order %s\n%s', recv_order, message)

    if execution_type == BinanceFuturesExecutionType.NEW:
      if not order.accepted:
        self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
      else:
        if order.internal.exchange_order_id != exchange_order_id:
          self._logger.error('Wrong exchange order id!\n%s\n%s', order, recv_order)
    elif execution_type == BinanceFuturesExecutionType.CANCELED:
      if self._order_manager.is_dead_order(order):
        self._logger.debug('Order (%s) is dead already, no need to set cancel_confirmed!',
                           order.order_id)
      else:
        self._order_manager.set_order(order, cancel_confirmed=True)
    elif execution_type == BinanceFuturesExecutionType.REJECTED:
      self._logger.info('Order rejected!, %s, %s', order, message)
      if order in self._order_manager:
        self._order_manager.remove_order(order, ignore_error=True)
    elif execution_type in (BinanceFuturesExecutionType.TRADE,
                            BinanceFuturesOrderStatus.FILLED,
                            BinanceFuturesOrderStatus.PARTIALLY_FILLED):
      if execution_type != BinanceFuturesExecutionType.TRADE:
        self._logger.info('execution_type: %s', message)
      last_fill = order.internal.fill.update_by_accum_fill(
          recv_order.internal.fill.value.price_avg_fill,
          recv_order.internal.fill.value.qty_accum_fill)
      if order.internal.fill.fully_filled:
        self._order_manager.set_order(order, fully_filled=True)

      if last_fill is None:
        self._logger.warning('Weird order status! %s, %s', json.dumps(message), order)
        return

      fill_id = '%s_%s' % (order.product.symbol, trade_id)
      if last_fill:
        fill_proto = ProductFillElement(fill_id=fill_id,
                                        price=last_fill.price_last_fill,
                                        qty=last_fill.qty_last_fill,
                                        order_id=order.order_id,
                                        side=to_proto_order_side(order.order_side),
                                        exchange_order_id=exchange_order_id,
                                        symbol=order.product.symbol)
        self.process_fill(fill_proto, timestamp, order=order)
      else:
        self._logger.error('Filled qty more than the orig_qty: %s', order.order_id)
      self._processed_fills.add(fill_id)

    elif execution_type == BinanceFuturesExecutionType.REPLACED:
      self._logger.error('%s execution type is not used!', execution_type)
    elif execution_type == BinanceFuturesOrderStatus.CALCULATED:
      self._logger.error('%s execution type is not used!', execution_type)
    elif execution_type == BinanceFuturesOrderStatus.EXPIRED:
      self._logger.error('%s execution type is not used!', execution_type)
    elif execution_type == BinanceFuturesOrderStatus.RESTATED:
      self._logger.error('%s execution type is not used!', execution_type)
    else:
      self._logger.error('Invalid execution_type: %s', execution_type)
